package ws.very.util.akka.cluster

import ws.very.util.akka.msg.IMsgCenter
import akka.cluster.ClusterEvent.MemberUp
import ws.very.util.akka.msg.MsgCenter
import akka.actor.RootActorPath
import akka.actor.ActorSystem
import akka.cluster.Cluster
import akka.actor.Props
import scala.concurrent.Promise
import akka.actor.ActorRef
import akka.pattern.ask
import akka.cluster.ClusterEvent.CurrentClusterState
import akka.actor.Stash
import akka.cluster.Member
import akka.actor.Address
import scala.util.Try
import akka.actor.Deploy
import akka.remote.RemoteScope
import akka.cluster.ClusterEvent.LeaderChanged

object ClusterMsgCenter {
  @deprecated
  case class SyncRegForward(reg: MsgCenter.RegForward)

  def apply(implicit sys: ActorSystem) =
    (sys.actorOf(Props[ClusterMsgCenter2], MsgCenter.name))
}
@deprecated
trait IClusterMsgCenter extends IMsgCenter with ClusterListener with Stash {
  def listen = Seq(classOf[MemberUp], classOf[LeaderChanged])
  import MsgCenter._
  //FIXME:不同步
  var leader: O[Address] = None
  def isSelfLeader = leader.exists { leaderAddr => cluster.selfAddress == leaderAddr }

  var members: Set[Member] = _
  override def receive = {
    case c: CurrentClusterState =>
      members = c.members
    case LeaderChanged(None) =>
      leader = None
      context.unbecome()

    case LeaderChanged(Some(newLeader)) =>
      leader = newLeader
      unstashAll()
      context.become(clusterListen.orElse { super.baseReceive })
    case other => stash
  }
  def clusterListen: Pfn[A, U] = {

    case MemberUp(member) =>
      members = members + member
      log.info(s"MemberUp member=$member cur=$curRegForwards")
      if ( /* isSelfLeader &&*/ member.address != cluster.selfAddress) {
        curRegForwards.foreach { reg =>
          remoteRefSend(member.address, ClusterMsgCenter.SyncRegForward(reg))
        }
      }
    case ClusterMsgCenter.SyncRegForward(reg) =>
      log.info(s"listen reg=$reg")
      if (rightReg(reg)) {
        super.add2Route(reg)
        if (isSelfLeader)
          members /*FIXME:不同步*/ .filter { member => member.address != cluster.selfAddress }.map { _.address }.foreach { add =>
            remoteRefSend(add, ClusterMsgCenter.SyncRegForward(reg))
          }
      }
  }

  def remoteRefSend(addr: Address, msg: A) = {
    log.info(s"send 2 $addr $msg")
    context.actorSelection(RootActorPath(addr) / "user" / name) ! msg
  }

  override def add2Route(reg: RegForward) = {
    log.info(s"add2Route members=$members reg=$reg")
    super.add2Route(reg)
    val t =
      (if (isSelfLeader)
        members /*FIXME:不同步*/ .filter { member => member.address != cluster.selfAddress }.map { _.address }
      else cluster.state.leader.toList)
    //        .filter { _ != sender().path.address }
    t.foreach { add =>
      remoteRefSend(add, ClusterMsgCenter.SyncRegForward(reg))
    }
  }

}
//集群前启动
@deprecated
protected class ClusterMsgCenter extends IClusterMsgCenter

//FIXME:leader不可替代master 可能迁移leader case Event(LeaderChanged(Some(addr)), curData)时同步自身?
protected class ClusterMsgCenter2 extends ClusterReplication with IMsgCenter {
  import MsgCenter._
  def addrRef(addr: Address) = context.actorSelection(RootActorPath(addr) / "user" / name)
  
  override def receive=({
    case any=>
      log.info(s"receive=$any")
      any
  }:Pfn[Any,Any]).andThen{super.receive}
  
  override def whenReady = ({
    case e @ Event(MemberUp(m), (Some(leader), members)) =>//override
      if (isSelfLeader(leader) && m.address != cluster.selfAddress) {
        val ref = addrRef(m.address)
        curRegForwards.foreach { r => ref ! r }
      }
      super.whenReady(e)
  }: StateFunction).orElse(super.whenReady).orElse {//biz
    case e @ Event(otherMsg, (Some(leader), members)) =>
      baseReceive(otherMsg)
      stay
  }

}